

namespace Stee.CAP8.MessageBus.Kafka.Factories
{
    using System;
    using Confluent.Kafka;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Factory to create new Kafka subscriber
    /// </summary>
    public static class SubscriberFactory
    {
        /// <summary>
        /// Create new subscriber
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <param name="subscriberGroup">Subscriber/consumer group. If specified,
        /// subscriber group will override group ID in JSON configuration.</param>
        /// <param name="isAutoCommit">Flag indicating if subscriber will auto commit message.
        /// If defined, flag will override configuration file.</param>
        /// <param name="subscriberErrorHandler">Subscriber error handler</param>
        /// <returns>Return new subscriber/consumer instance</returns>
        public static IConsumer<string, byte[]> CreateNewSubscriber(string jsonConfig, string subscriberGroup,
            bool? isAutoCommit, Action<IConsumer<string, byte[]>, Error> subscriberErrorHandler)
        {
            var handler = subscriberErrorHandler ?? throw new ArgumentNullException(nameof(subscriberErrorHandler));
            var config = InitializeSubscriberConfig(jsonConfig, subscriberGroup, isAutoCommit);
            var client = BuildSubscriber(config, handler);

            return client;
        }


        #region Private Methods

        /// <summary>
        /// Initialize subscriber configuration
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <param name="subscriberGroup">Subscriber/consumer group.
        /// If specified, subscriber group will override group ID in JSON configuration</param>
        /// <param name="isAutoCommit">Flag indicating if subscriber will auto commit message.
        /// If defined, flag will override configuration file.</param>
        /// <returns>Return new subscriber/consumer configuration</returns>
        private static ConsumerConfig InitializeSubscriberConfig(string jsonConfig, string subscriberGroup, bool? isAutoCommit)
        {
            var configs = ConfigurationHelper.GetBrokerConfiguration(jsonConfig, ConfigurationConstants.BrokerConfigSubscribeSection);
            var configuration = new SubscriberConfiguration(configs);
            var group = ValidateHelper.CheckString(subscriberGroup, excpMsg: "Subscriber group ID cannot be null or empty.");

            return new ConsumerConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                GroupId = group,
                ClientId = configuration.ClientID,
                AutoOffsetReset = configuration.AutoOffsetReset,
                EnableAutoCommit = isAutoCommit ?? configuration.EnableAutoCommit,
                EnableAutoOffsetStore = configuration.EnableAutoOffsetStore,
                MaxPollIntervalMs = configuration.MaxPollIntervalMs,
                SessionTimeoutMs = configuration.SessionTimeoutMs,
                AutoCommitIntervalMs = configuration.AutoCommitIntervalMs,
                CheckCrcs = configuration.CheckCrcs,
                PartitionAssignmentStrategy = configuration.PartitionAssignmentStrategy,
            };
        }

        /// <summary>
        /// Build subscriber
        /// </summary>
        /// <param name="consumerConfig">Consumer configuration</param>
        /// <param name="subscriberErrorHandler">Consumer error handler</param>
        /// <returns>Return new subscriber/consumer instance</returns>
        private static IConsumer<string, byte[]> BuildSubscriber(ConsumerConfig consumerConfig, Action<IConsumer<string, byte[]>, Error> subscriberErrorHandler)
        {
            //// Note: If a key or value deserializer is not set, the 
            //// deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
            //// will be used automatically (where available). The default deserializer for string
            //// is UTF8. The default deserializer for Ignore returns null for all input data (including non-null data).
            var consumerBuilder = new ConsumerBuilder<string, byte[]>(consumerConfig);
            consumerBuilder.SetKeyDeserializer(Deserializers.Utf8);
            consumerBuilder.SetValueDeserializer(Deserializers.ByteArray);
            consumerBuilder.SetErrorHandler(subscriberErrorHandler);

            return consumerBuilder.Build();
        }

        #endregion Private Methods
    }
}
